package com.bw.gmall.realtime.dim.functions;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.bean.TableProcessDim;
import com.bw.gmall.realtime.common.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.List;

public class DimBroadcastFunction extends BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>> {
    // 广播流和主流没有先后顺序之分
    private HashMap<String, TableProcessDim> hashMap;
    private MapStateDescriptor<String, TableProcessDim> stateDescriptor;

    public DimBroadcastFunction(MapStateDescriptor<String, TableProcessDim> stateDescriptor) {
        this.stateDescriptor = stateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        hashMap = new HashMap<>();
        // 查一下mysql,放到内存中
        java.sql.Connection mysqlConnection = JdbcUtil.getMysqlConnection();
        List<TableProcessDim> tableProcessDims = JdbcUtil.queryList(mysqlConnection, "select * from gmall2023_config.table_process_dim", TableProcessDim.class, true);
        for (TableProcessDim tableProcessDim : tableProcessDims) {
            hashMap.put(tableProcessDim.getSourceTable(), tableProcessDim);
        }
        JdbcUtil.closeConnection(mysqlConnection);
    }

    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = readOnlyContext.getBroadcastState(stateDescriptor);
        // 取这个souretable
        String table = jsonObject.getString("table");
        // 取状态值
        TableProcessDim tableProcessDim = broadcastState.get(table);
        if (tableProcessDim == null) {
            // 内存查找
            tableProcessDim = hashMap.get(table);
        }

        if (tableProcessDim != null) {
            collector.collect(Tuple2.of(jsonObject, tableProcessDim));
        }
    }

    @Override
    public void processBroadcastElement(TableProcessDim tableProcessDim, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context context, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        BroadcastState<String, TableProcessDim> broadcastState = context.getBroadcastState(stateDescriptor);
        String sourceTable = tableProcessDim.getSourceTable();
        String op = tableProcessDim.getOp();
        if ("d".equals(op)) {
            // 从内存也要删除掉
            hashMap.remove(sourceTable);
            // 从状态删除掉
            broadcastState.remove(sourceTable);
        } else {
            broadcastState.put(sourceTable, tableProcessDim);
        }

    }
}
